import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.Iterator;

public class PiEstimator extends Configured implements Tool {
    /**
     * 临时目录的路径，保存运行中文件的数据
     */
    static private final Path TMP_DIR = new Path(PiEstimator.class.getSimpleName() + "_TMP_3_1415926");

    /**
     * Halton序列类，产生随机点
     */
    private static class HaltonSequence {
        static final int[] P = {2, 3};
        static final int[] K = {63, 64};

        private long index;
        private double[] x;
        private double[][] q;
        private int[][] d;

        /**
         * 构造函数
         *
         * @param startIndex
         * @return
         */
        HaltonSequence(long startIndex) {
            index = startIndex;
            x = new double[K.length];
            q = new double[K.length][];
            d = new int[K.length][];
            for (int i = 0; i < K.length; i++) {
                q[i] = new double[K[i]];
                d[i] = new int[K[i]];
            }
            for (int i = 0; i < K.length; i++) {
                long k = index;
                x[i] = 0;
                for (int j = 0; j < K[i]; j++) {
                    q[i][j] = (j == 0 ? 1.0 : q[i][j - 1]) / P[i];
                    d[i][j] = (int) (k % P[i]);
                    k = (k - d[i][j]) / P[i];
                    x[i] += d[i][j] * q[i][j];
                }
            }
        }

        /**
         * 生成点
         */
        double[] nextPoint() {
            index++;
            for (int i = 0; i < K.length; i++) {
                for (int j = 0; j < K[i]; j++) {
                    d[i][j]++;
                    x[i] += q[i][j];
                    if (d[i][j] < P[i]) {
                        break;
                    }
                    d[i][j] = 0;
                    x[i] -= (j == 0 ? 1.0 : q[i][j - 1]);
                }
            }
            return x;
        }
    }

    /**
     * 定义Map过程
     *
     * @author Peng
     * @date 2018/12/4
     */
    public static class PiMapper extends MapReduceBase implements Mapper<LongWritable, LongWritable, BooleanWritable, LongWritable> {
        @Override
        public void map(LongWritable offset, LongWritable size, OutputCollector<BooleanWritable, LongWritable> out, Reporter reporter) throws IOException {

            /**
             * 函数从序列文件里面得到offset和size， offset是要产生的而言本点在Halton序列里的序号，size是这个Map任务需要产生的样本点数量
             */
            final HaltonSequence haltonSequence = new HaltonSequence(offset.get());
            long numInside = 0;
            long numOutside = 0;
            for (long i = 0; i < size.get(); ) {
                // 生成随机点
                final double[] point = haltonSequence.nextPoint();
                // 判断随机点是否在扇形内
                final double x = point[0] - 0.5;
                final double y = point[1] - 0.5;
                if (x * x + y * y > 0.25) {
                    numOutside++;
                } else {
                    numInside++;
                }
                i++;
                if (i % 1000 == 0) {
                    reporter.setStatus("Generator " + i + "samples.");
                }
            }
            out.collect(new BooleanWritable(true), new LongWritable(numInside));
            out.collect(new BooleanWritable(false), new LongWritable(numOutside));
        }
    }

    /**
     * PiReduce类，定义Reduce过程
     *
     * @author Peng
     * @date 2018/12/4
     */
    public static class PiReduce extends MapReduceBase implements Reducer<BooleanWritable, LongWritable, WritableComparable<?>, Writable> {
        private long numInside = 0;
        private long numOutside = 0;
        private JobConf conf;

        @Override
        public void configure(JobConf job) {
            conf = job;
        }

        @Override
        public void reduce(BooleanWritable isInside,
                           Iterator<LongWritable> values,
                           OutputCollector<WritableComparable<?>, Writable> output,
                           Reporter reporter) throws IOException {
            /**
             * isInside是布尔类型值，true或者false
             * values包含10个数值，分表来自10个Map任务
             */

            if (isInside.get()) {
                while (values.hasNext()) {
                    numInside += values.next().get();
                }
            } else {
                while (values.hasNext()) {
                    numOutside += values.next().get();
                }
            }
        }

        /**
         * Reduce过程结束后，将计算结果写到临时目录的文件中，以供PiEstimator类读取
         *
         * @author Peng
         * @date 2018/12/4
         */
        @Override
        public void close() throws IOException {
            Path outDir = new Path(TMP_DIR, "out");
            Path outFile = new Path(outDir, "reduce-out");
            FileSystem fileSys = FileSystem.get(conf);
            SequenceFile.Writer writer = SequenceFile.createWriter(fileSys, conf, outFile, LongWritable.class, LongWritable.class, SequenceFile.CompressionType.NONE);
            writer.append(new LongWritable(numInside), new LongWritable(numOutside));
            writer.close();
        }
    }

    /**
     * PiEstimator类的estimate函数，运行MapReduce，然后计算Pi值
     *
     * @author Peng
     * @date 2018/12/4
     */
    private static BigDecimal estimate(int numMaps, long numPoints, JobConf jobConf) throws IOException {
        jobConf.setJobName(PiEstimator.class.getSimpleName());
        jobConf.setInputFormat(SequenceFileInputFormat.class);

        jobConf.setOutputKeyClass(BooleanWritable.class);
        jobConf.setOutputValueClass(LongWritable.class);
        jobConf.setOutputFormat(SequenceFileOutputFormat.class);

        jobConf.setMapperClass(PiMapper.class);
        jobConf.setNumMapTasks(numMaps);
        jobConf.setReducerClass(PiReduce.class);
        jobConf.setNumReduceTasks(1);

        /**
         * 关闭speculative execution功能
         * 该功能指的是Hadoop发现有些任务执行的比较慢的时候，会在其他节点上执行相同的任务，两个任务谁先执行完成就以谁为准
         * 但是Reduce写需要将数值写入到HDFS的文件中，该文件名是固定的，如果同时运行两个以上的Reduce任务，会导致写入出错，因此需要关闭此功能
         */
        jobConf.setSpeculativeExecution(false);

        /**
         * 设置输入目录和输出目录
         */
        final Path inDir = new Path(TMP_DIR, "in");
        final Path outDir = new Path(TMP_DIR, "out");
        FileInputFormat.setInputPaths(jobConf, inDir);
        FileOutputFormat.setOutputPath(jobConf, outDir);

        /**
         * 生成输入目录
         */
        final FileSystem fs = FileSystem.get(jobConf);

        if (fs.exists(TMP_DIR)) {
            throw new IOException("Tmp directory " + fs.makeQualified(TMP_DIR) + "already exists, Please remove it first.");
        }

        if (!fs.mkdirs(inDir)) {
            throw new IOException("Cannot create input directory " + inDir);
        }

        try {
            /**
             * 为每个Map任务生成一个序列文件，并写入数值
             */
            for (int i = 0; i < numMaps; ++i) {
                final Path file = new Path(inDir, "part" + i);
                final LongWritable offset = new LongWritable(i * numPoints);
                final LongWritable size = new LongWritable(numPoints);
                try (SequenceFile.Writer writer = SequenceFile.createWriter(
                        fs, jobConf, file, LongWritable.class, LongWritable.class, SequenceFile.CompressionType.NONE)) {
                    writer.append(offset, size);
                }
                System.out.println("Wrote input for Map #" + i);
            }

            /**
             * 运行MapReduce任务
             */
            System.out.println("Starting job");
            final long startTime = System.currentTimeMillis();
            JobClient.runJob(jobConf);
            final double duration = (System.currentTimeMillis() - startTime) / 1000.0;
            System.out.println("Job finished in " + duration + "seconds");

            /**
             * 读取输出结果
             */
            Path inFile = new Path(outDir, "reduce-out");
            LongWritable numInside = new LongWritable();
            LongWritable numOutSide = new LongWritable();
            SequenceFile.Reader reader = new SequenceFile.Reader(fs, inFile, jobConf);
            try {
                reader.next(numInside, numOutSide);
            } finally {
                reader.close();
            }

            /**
             * 计算Pi值然后返回
             */
            return BigDecimal.valueOf(4).setScale(20)
                    .multiply(BigDecimal.valueOf(numInside.get()))
                    .divide(BigDecimal.valueOf(numMaps))
                    .divide(BigDecimal.valueOf(numPoints));
        } finally {
            fs.delete(TMP_DIR, true);
        }
    }

    /**
     * Tool接口的run函数
     *
     * @author Peng
     * @date 2018/12/4
     */
    @Override
    public int run(String[] strings) throws Exception {
        if (strings.length != 2) {
            System.out.println("Usage: " + getClass().getName() + "<nMaps> <nSamples>");
            ToolRunner.printGenericCommandUsage(System.err);
            return -1;
        }

        final int nMaps = Integer.parseInt(strings[0]);
        final long nSamples = Integer.parseInt(strings[1]);

        System.out.println("Number of Maps = " + nMaps);
        System.out.println("Samples per Map = " + nSamples);

        final JobConf jobConf = new JobConf(getConf(), getClass());
        System.out.println("Estimated value of Pi is " + estimate(nMaps, nSamples, jobConf));
        return 0;
    }

    /**
     * PiEstimator类的main函数
     *
     * @author Peng
     * @date 2018/12/4
     */
    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(null, new PiEstimator(), args));
    }
}
